package org.zjt.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.zjt.demo.util.HBaseHelper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * DESC
 *
 * @author
 * @create 2017-04-19 下午9:45
 **/
public class TestBigDataInsert {
    private static Connection connection;
    private static String tableName = "datatable";
    private static String colfam = "colfam";

    static {
        try {
            connection = getConnetcion(tableName, colfam);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static Connection getConnetcion(String table, String colfam) throws Exception {
        Configuration conf = HBaseConfiguration.create(); // co GetExample-1-CreateConf Create the configuration.
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        // ^^ GetExample
        HBaseHelper helper = HBaseHelper.getHelper(conf);
        if (!helper.existsTable(table)) {
            helper.createTable(table, colfam);
        }
        // vv GetExample
        return ConnectionFactory.createConnection(conf);
    }


    public static void closeConnetcion(Connection connection) throws Exception {
        if (connection != null)
            connection.close();
    }


    public static void main(String[] args)throws Exception{


        testScan();
    }



    /**
     * 性能：1090000条数据查询出10000需要0.3秒
     * @throws Exception
     */
    public static void testScan() throws Exception {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(colfam));
        /**
         * 可添加多个column family：
         * scan.addFamily(Bytes.toBytes(colfam));
         * */

        scan.setStartRow(Bytes.toBytes("row2"));
        scan.setMaxResultSize(0);
        Filter filter = new PageFilter(10000);
        scan.setFilter(filter);
        long start = System.currentTimeMillis();
        ResultScanner scanner = table.getScanner(scan);
        scanner.forEach(a -> Arrays.stream(a.rawCells()).map(cell -> "Cell: " + cell + ", Value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength())).forEach(System.out::println));

        System.out.println((System.currentTimeMillis() - start)/1000);
    }
    /***
     * 初始化数据
     * @throws Exception
     */
    public static void init()throws Exception{
        Table table = connection.getTable(TableName.valueOf(tableName));
        for (int j = 0; j < 100; j++) {
            List<Row> rows = new ArrayList<>();
            String[] quals = {"qual1", "age", "qq"};
            for (int i = 0; i < 100_000; i++) {
                for (String qual : quals){
                    try {
                        Put put = new Put(Bytes.toBytes("row"+(i+10_000*j)));
                        put.add(new KeyValue(Bytes.toBytes("row"+(i+10_000*j)), Bytes.toBytes(colfam), Bytes.toBytes(qual), Bytes.toBytes("row"+(i+10_000*j)+qual)));
                        rows.add(put);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }

            Object[] resultArray = new Object[rows.size()];
            /** 异常也能得到返回结果 */
            table.batch(rows, resultArray);
            Arrays.stream(resultArray).forEach(a -> {
                // TODO: 2017/4/19 执行get得到的结果处理
                System.out.println(a);
            });
        }
        table.close();
        closeConnetcion(connection);
    }
}
